package com.example.rdddemo.controller;

import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.web.bind.annotation.*;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

@RestController
@RequestMapping("/api/movies")
public class MovieController {

    private static final Logger logger = LoggerFactory.getLogger(MovieController.class);

    private final SparkSession sparkSession;
    private final JavaSparkContext sparkContext;

    public MovieController(SparkSession sparkSession, JavaSparkContext sparkContext) {
        this.sparkSession = sparkSession;
        this.sparkContext = sparkContext;
    }

    @GetMapping("/filter")
    public Map<String, Object> filterMovies(@RequestParam String keyword) {
        logger.info("关键字: {}", keyword);

        // 读取csv文件
        Dataset<Row> movies = sparkSession.read()
                .option("header", "true")
                .csv("hdfs://192.168.5.128:9000/tmp/movies.csv");

        logger.info("CSV文件加载成功。行数: {}", movies.count());

        // 将DataFrame转换为RDD
        JavaRDD<Row> movieRDD = movies.toJavaRDD();

        // 过滤字段
        JavaRDD<Row> filteredRDD = movieRDD.filter(row -> {
            boolean containsKeyword = row.getAs("title").toString().contains(keyword);
            if (containsKeyword) {
                logger.debug("找到匹配的电影: {}", (Object) row.getAs("title"));
            }
            return containsKeyword;
        });

        logger.info("过滤完成。匹配行数: {}", filteredRDD.count());

        // 映射数据
        List<Map<String, Object>> movieList = filteredRDD.map(row -> {
            Map<String, Object> movieMap = new HashMap<>();
            movieMap.put("movieId", row.getAs("movieId"));
            movieMap.put("title", row.getAs("title"));
            movieMap.put("genres", row.getAs("genres"));
            logger.debug("电影结果: {}", (Object) movieMap);
            return movieMap;
        }).collect();

        logger.info("映射完成。结果中的电影数量: {}", movieList.size());

        // 创建响应
        Map<String, Object> response = new HashMap<>();
        if (movieList.isEmpty()) {
            response.put("message", "没有找到匹配的电影。");
            response.put("movies", movieList);
        } else {
            response.put("movies", movieList);
        }

        return response;
    }
}
